from mrjob.job import MRJob
from mrjob.step import MRStep
import re


WORD_RE = re.compile(r"[\w']+")


class MRWordCount(MRJob):
    """
    一步任务
    """
    def mapper(self, key, line):
        """
        mapper:接收每一行的输入，处理后返回一堆dict
        :param key:
        :param line:
        :return:
        """
        for word in line.split():
            yield word, 1

    def reducer(self, word, occurrences):
        """
        reduceer：接收mapper的dict，对其进行整合，把相同key的value累加后输出
        :param word:
        :param occurrences:
        :return:
        """
        yield word, sum(occurrences)


class MRMultiWordCount(MRJob):
    """
    多步任务：代码来源：https://www.jianshu.com/p/21c880ee93a9
    """
    def steps(self):
        """
        覆盖steps方法，并在step中返回一个由mapper, combiner, reducer等组成的list
        :return:
        """
        return [MRStep(mapper=self.mapper_get_words,
                       combiner=self.combiner_count_words,
                       reducer=self.reducer_count_words),
                MRStep(reducer=self.reducer_find_max_word)]

    def mapper_get_words(self, _, line):
        for word in WORD_RE.findall(line):
            yield (word.lower(), 1)

    def combiner_count_words(self, word, counts):
        yield (word, sum(counts))

    def reducer_count_words(self, word, counts):
        yield None, (sum(counts), word)

    def reducer_find_max_word(self, _, word_count_pairs):
        yield max(word_count_pairs)


if __name__ == '__main__':
    MRWordCount.run()
